/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.splitor;

import cn.ac.iie.di.tools.FileTools;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Calendar;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.logging.Level;
import org.apache.log4j.Logger;

/**
 *
 * @author lenovo-pc
 */
public class JsonSplitor implements Runnable {

    String dir = null;
    String desdir = null;
    ArrayBlockingQueue dataQueue = null;
    String FileString = null;
    Long blockingQSize = null;
    static Logger logger = null;

    static {
        logger = org.apache.log4j.Logger.getLogger(JsonSplitor.class.getName());
    }

    public JsonSplitor(String dir, String desdir, String filetype, ArrayBlockingQueue dataQueue) {
        this.blockingQSize = 0L;
        this.dir = dir;
        this.dataQueue = dataQueue;
        this.desdir = desdir;
        this.FileString = filetype;
    }

    public void run() {
        while (true) {
//            logger.info("the json parser dir is " + dir);
            List<File> sigFiles = FileTools.getFiles(new File(dir), "log");

//            logger.info("current ok files " + sigFiles);

            for (File sigFile : sigFiles) {
                File filelock = new File(sigFile + ".lock");
                if (!filelock.exists()) {
                    try {
                        String dataFilePath = sigFile.getPath();
//                        logger.info("file:" + dataFilePath);
                        if (!FileTools.creatTxtFile(dataFilePath + ".lock")) {
//                            logger.info(".lock Create failed :" + dataFilePath);
                        }
//                        sigFile.getName().hashCode()%3
                        readFileByLines(dataFilePath, desdir, FileString);
                    } catch (IOException ex) {
                        logger.error(".lock Create error !" + ex, ex);
                    }
                }
            }
//            try {
//                Thread.sleep(3000);
//            } catch (InterruptedException ex) {
//            }
        }
    }

    public void readFileByLines(String fileName, String newFileName, String FileType) {

        BufferedReader reader = null;
        String filetype = null;
        filetype = FileType;
        try {

            long begin = System.currentTimeMillis();
            //System.out.println("以行为单位读取文件内容，一次读一整行：");
            //reader = new BufferedReader(new FileReader(fileName));
            reader = new BufferedReader(new InputStreamReader(new FileInputStream(fileName), "UTF-8"));
            String tempString = null;
            int line = 0;
            // 一次读入一行，直到读入null为文件结束  

            while ((tempString = reader.readLine()) != null) {
                this.blockingQSize++;
                if (blockingQSize > 1000) {
                    logger.debug("ArrayBlockingQueue Size is " + dataQueue.remainingCapacity());
                    blockingQSize = 0L;
                }
//                try {
                while (!dataQueue.offer(tempString)) {
//                        Thread.sleep(1000);
//                        logger.warn("Put Json Array Block buffer Fail!");
                }

//                } catch (InterruptedException ex) {
//                }
                line++;
            }
            reader.close();
//            if (!FileTools.creatTxtFile(fileName + ".lock")) {
//                logger.info(".lock Create failed :" + fileName);
//            }

            logger.info("There is " + line + " in " + fileName + "！ ");
            //将文件移动走，删除.ok文件和解压文件，移动.gz和.sig文件

            FileTools.moveFileToNewPath(fileName, newFileName);
            long end = System.currentTimeMillis();
            System.out.println("consume " + (end - begin) + " ms");
        } catch (Exception e) {
            logger.error("the error is :" + e, e);
        }
    }
}
